package org.example;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.scala.Serdes;
import org.junit.Test;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * Hello world!
 */
public class App {
    public static void main(String[] args) {
        System.out.println("Hello World!");
    }


    @Test
    public void testProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");

        //ack是判别请求是否为完整的条件（就是是判断是不是成功发送了）。我们指定了“all”将会阻塞消息，这种设置性能最低，但是是最可靠的。
        props.put("acks", "all");

        //retries，如果请求失败，生产者会自动重试，我们指定是0次，如果启用重试，则会有重复消息的可能性。
        props.put("retries", 0);

        //默认缓冲可立即发送，即便缓冲空间还没有满，但是，如果你想减少请求的数量，可以设置linger.ms大于0。这将指示生产者发送请求之前等待
        //一段时间，希望更多的消息填补到未满的批中。这类似于TCP的算法，例如上面的代码段，可能100条消息在一个请求发送，因为我们设置了linger(逗留)
        //时间为1毫秒，然后，如果我们没有填满缓冲区，这个设置将增加1毫秒的延迟请求以等待更多的消息。需要注意的是，在高负载下，相近的时间一般也会组
        //成批，即使是 linger.ms=0。在不处于高负载的情况下，如果设置比0大，以少量的延迟代价换取更少的，更有效的请求
        props.put("linger.ms", 1);

        //单次批量的最大数据数，如果大于该值，不用调用producer.close，就会将消息发送出去
        props.put("batch.size", 16384);

        //buffer.memory 控制生产者可用的缓存总量，如果消息发送速度比其传输到服务器的快，将会耗尽这个缓存空间。
        // 当缓存空间耗尽，其他发送调用将被阻塞，阻塞时间的阈值通过max.block.ms设定，之后它将抛出一个TimeoutException
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //producer(生产者)缓存每个分区未发送的消息。缓存的大小是通过 batch.size 配置指定的。值较大的话将会产生更大的批。并需要更多的内存（因为每个“活跃”的分区都有1个缓冲区）。
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 5; i++) {
            //send()方法是异步的，添加消息到缓冲区等待发送，并立即返回。生产者将单个的消息批量在一起发送来提高效率
            producer.send(new ProducerRecord<String, String>("test", "Key_" + Integer.toString(i), "value_" + Integer.toString(i)));
        }

        //如果不调用该方法，就没法发送消息到队列
        //生产者的缓冲空间池保留尚未发送到服务器的消息，后台I/O线程负责将这些消息转换成请求发送到集群。如果使用后不关闭生产者，则会丢失这些消息。
        producer.close();
    }


    @Test
    public void testProducer1() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 3; i++) {
            //异步回调
            producer.send(new ProducerRecord<String, String>("test", "key_a", "value_a"), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e != null)
                        e.getStackTrace();
                    System.out.println(recordMetadata.hasOffset() + ":"
                            + recordMetadata.hasTimestamp() + ":"
                            + recordMetadata.offset() + ":"
                            + recordMetadata.partition() + ":"
                            + recordMetadata.serializedKeySize() + ":"
                            + recordMetadata.serializedValueSize() + ":"
                            + recordMetadata.timestamp() + ":"
                            + recordMetadata.topic() + ":"
                            + recordMetadata);
                }
            });
            System.out.println("i="+i);
        }

        producer.close();
    }

    @Test
    public void testProducer2() throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        //同步回调
        for (int i = 0; i < 3; i++) {
            RecordMetadata recordMetadata = producer.send(new ProducerRecord<String, String>("test", 0, "key_" + i, "value_" + i)).get();
            System.out.println(recordMetadata.hasOffset() + ":"
                    + recordMetadata.hasTimestamp() + ":"
                    + recordMetadata.offset() + ":"
                    + recordMetadata.partition() + ":"
                    + recordMetadata.serializedKeySize() + ":"
                    + recordMetadata.serializedValueSize() + ":"
                    + recordMetadata.timestamp() + ":"
                    + recordMetadata.topic() + ":"
                    + recordMetadata);
        }
        producer.close();
    }



    @Test
    public void testConsumer(){
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test");
        //enable.auto.commit cannot be set to true when default group id (null) is used.
        //props.setProperty("enable.auto.commit", "true");



        //1.Consumer消费了5条消息后宕机了，重启之后它读取到对应的partition的Committed Offset为5，因此会直接从第6条消息开始读取。此时完全依赖于Committed Offset机制，和auto.offset.reset配置完全无关。
        //2.新建了一个新的Group，并添加了一个Consumer，它订阅了一个已经存在的Topic。此时Kafka中还没有这个Consumer相应的Offset信息，因此此时Kafka就会根据auto.offset.reset配置来决定这个Consumer从何处开始消费消息。
        //从头开始消费
        //props.setProperty("auto.offset.reset", "earliest");
        //消费该分区下最新产生的消息
        //props.setProperty("auto.offset.reset", "latest");
        //props.setProperty("auto.offset.reset", "none");

        //max.poll.interval.ms：增大poll的间隔，可以为消费者提供更多的时间去处理返回的消息（调用poll(long)返回的消息，通常返回的消息都是一批）。缺点是此值越大将会延迟组重新平衡。
        props.setProperty("auto.commit.interval.ms", "10000");

        //max.poll.records：此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔，减少重新平衡分组的
        props.setProperty("max.poll.records", "10");

        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test"));//Arrays.asList("foo", "bar")
        while (true) {
            //推荐方法是将消息处理移到另一个线程中。 poll（5000）：即在5s中内拉去的数据返回到消费者端。
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records){
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            //同步，手动设置已经消费
            //consumer.commitSync();
            //异步，
            consumer.commitAsync();
        }
    }


    public void testStream(){
        
    }
}
